package com.frp.lib.handler;

import com.frp.lib.client.Proxy;
import com.frp.lib.utils.NettyUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by hujianjun on 12/6 0006.
 */
public class ForwardHandler extends ChannelInboundHandlerAdapter {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    private volatile Channel outboundChannel;
    private final Proxy proxy;
    private int inBytes = 0;
    private int outBytes = 0;

    public ForwardHandler(Proxy proxy) {
        this.proxy = proxy;
    }

    public void initLocal(ChannelHandlerContext ctx) {
        final Channel inboundChannel = ctx.channel();
        this.logger = LoggerFactory.getLogger("forward-" + proxy.getProxyDefine().getName() + "-" + ctx.channel().localAddress());
        final String host = proxy.getProxyDefine().getLocal_ip();
        final int port = proxy.getProxyDefine().getLocal_port();
        logger.info("begin to init local connect {}:{}", host, port);
        ChannelHandler handler = new LocalHandler(this, proxy.getProxyDefine().getName(), inboundChannel);
//        ChannelHandler logHandler = new LoggingHandler("local-" + proxy.getProxyDefine().getName() + "-" + inboundChannel.localAddress());
        this.outboundChannel = NettyUtils.connectLocal(host, port, handler);
        logger.info("connect local success:{}", this.outboundChannel);

        int timeWait = 0;
        while (!outboundChannel.isActive()) {
            try {
                Thread.sleep(100);
                timeWait += 100;
                if (timeWait > 5000) {
                    //最多等待5秒
                    logger.error("local connect error.");
                    break;
                }
            } catch (InterruptedException e) {
                logger.error("sleep error.", e);
            }
        }
        logger.info("tunnel created for {}", outboundChannel);
    }

    @Override
    public void channelRead(final ChannelHandlerContext ctx, Object msg) {
        int timeWait = 0;
        while(outboundChannel==null || !outboundChannel.isActive()) {
        try {
            Thread.sleep(100);
            timeWait += 100;
            if (timeWait > 7000) {
                //最多等待7秒
                logger.error("local connect error.");
                break;
            }
        } catch (InterruptedException e) {
            logger.error("sleep error.", e);
        }
        }
//        if (null == outboundChannel) {
//            initLocal(ctx);
//        }
        if (msg instanceof ByteBuf) {
            ByteBuf bf = (ByteBuf)msg;
//            String data = ByteBufUtil.hexDump(bf);
//            logger.info("forward remote read:{}", data);
            inBytes += bf.readableBytes();
        }
        if (outboundChannel.isActive()) {
            ReferenceCountUtil.retain(msg); //避免write后计数器为0
            outboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {
                    if (future.isSuccess()) {
                        // was able to flush out data, start to read the next chunk
                        ctx.channel().read();
                    } else {
                        future.channel().close();
                    }
                }
            });
            try {
                super.channelRead(ctx, msg);    //继续处理数据
            } catch (Exception e) {
                logger.error("read error.", e);
            }
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        logger.info("tunnel closed for {}, in:{} bytes, out:{} bytes", outboundChannel, inBytes, outBytes);
        this.proxy.removeChannelMap(ctx.channel());
        if (outboundChannel != null) {
            closeOnFlush(outboundChannel);
        }
        try {
            super.channelInactive(ctx);     //继续处理消息
        } catch (Exception e) {
            logger.error("inactive error.", e);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error("exception.", cause);
        closeOnFlush(ctx.channel());
    }

    public void incOutBytes(int size) {
        outBytes += size;
    }

    /**
     * Closes the specified channel after all queued write requests are flushed.
     */
    static void closeOnFlush(Channel ch) {
        if (ch.isActive()) {
            ch.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
        }
    }

    static class LocalHandler extends ChannelInboundHandlerAdapter {
        private final Logger logger;
        private final Channel inboundChannel;
        private final String name;
        private ForwardHandler handler;

        LocalHandler(ForwardHandler handler, String name, Channel inboundChannel) {
            this.handler = handler;
            this.name = name;
            this.inboundChannel = inboundChannel;
            logger = LoggerFactory.getLogger("local-" + name + "-" + inboundChannel.localAddress());
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            ctx.read();
        }

        @Override
        public void channelRead(final ChannelHandlerContext ctx, Object msg) {
            if (msg instanceof ByteBuf) {
                ByteBuf bf = (ByteBuf)msg;
                handler.incOutBytes(bf.readableBytes());
//                ForwardHandler.this.outBytes += bf.readableBytes();
//                logger.info("forward local read:{}", ByteBufUtil.hexDump(data));
            }
            inboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {
                    if (future.isSuccess()) {
                        ctx.channel().read();
                    } else {
                        logger.error("remote write error!");
                        future.channel().close();
                    }
                }
            });
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            closeOnFlush(inboundChannel);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            logger.error("exception", cause);
            closeOnFlush(ctx.channel());
        }

    }

}
